package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: kongcb
 * @Description: TODO
 * @Date: 2021/12/15 9:19
 */
public class TEciBranch {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_eci_branch").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_branch");
        long zs_count = zs_data.count();
        if(zs_count>0){
            zs_data.createOrReplaceTempView("zs_t_branch_tmp");
            //分支机构表获取企业（主体）相应的统一社会信用代码或者名称，用于后续关联取company_id
            Dataset<Row> zs_branch_del = spark.sql("select a.*,b.name " +
                    "from a_dc_ep_ods.zs_t_branch_del a " +
                    "left join " +
                    "a_dc_ep_ods.com_company b on a.entid = b.entid " +
                    "left join zs_t_branch_tmp c on a.entid=c.entid " +
                    "where c.entid is null");
            zs_branch_del.createOrReplaceTempView("zs_branch_del");

            spark.sql("select a.*,b.creditcode,c.name,c.start_date as esdate, c.oper_name " +
                            "from  zs_t_branch_tmp a " +
                            "left join " +
                            "a_dc_ep_ods.zs_t_branch_del b " +
                            "on a.entid=b.entid " +
                            "left join " +
                            "a_dc_ep_ods.com_company c " +
                            "on a.entid=c.entid")
                    .createOrReplaceTempView("zs_branch");
            spark.catalog().dropTempView("zs_t_branch_tmp");
            //del:根据统一社会信用代码和企业名称过滤需要删除的分支机构数据
            Dataset<Row> qcc_branch_del = spark.sql("select " +
                    "a.id as record_id," +
                    "a.id," +
                    "a.key_no," +
                    "a.company_id," +
                    "a.company_name," +
                    "a.sub_key_no," +
                    "a.sub_company_id," +
                    "a.name," +
                    "a.reg_no," +
                    "a.belong_org," +
                    "a.dates," +
                    "'-1' as isadd " +
                    "from a_dc_ep_dwi.t_eci_branch a " +
                    "where " +
                    "a.company_name in " +
                    "(select d.name from zs_branch_del d where d.name is not null " +
                    "and d.name != '' and length(trim(d.creditcode))<9) " +
                    "or " +
                    "a.company_id in " +
                    "(select b.company_id from a_dc_ep_dwi.t_eci_company b " +
                    "where b.credit_code in " +
                    "(select c.creditcode from zs_branch_del c where length(trim(c.creditcode))>=9))");
            qcc_branch_del.createOrReplaceTempView("qcc_branch_del");
            spark.catalog().dropTempView("zs_branch_del");
            //updated && insert:将中数推送过来的数据全部映射到对应的企查查表字段。温馨提示：此处表的id可以设置成对应的中数的entid，如果中经社没有该企业数据则置空，后续如果中数推送了该企业数据可以通过该字段关联补充置空的数据。
            Dataset<Row> qcc_branch_update_code = spark.sql(" select " +
                    "a.record_id," +
                    "a.entid as id," +
                    "if(length(trim(d.credit_code))==18,md5(trim(d.credit_code)),concat(d.company_id,'#')) as key_no," +
                    "if(d.company_id is not null,d.company_id,md5(trim(a.creditcode))) as company_id," +
                    "if(d.company_id is not null,d.company_name,a.name) as company_name," +
                    "if(length(a.BRN_CREDIT_CODE)==18,md5(trim(a.BRN_CREDIT_CODE)),concat(md5(trim(a.BRNAME)),'#')) as sub_key_no," +
                    "a.BRNAME as name," +
                    "a.BRREGNO as reg_no," +
                    "a.BRN_REG_ORG as belong_org," +
                    "a.BRN_CREDIT_CODE," +
                    "a.jobid " +
                    "from (select * from zs_branch c where length(trim(c.creditcode)) =18) a " +
                    "left join " +
                    "(select b.company_id,b.company_name,b.credit_code from a_dc_ep_dwi.t_eci_company b where length(trim(b.credit_code)) =18) d  " +
                    "on trim(d.credit_code)=trim(a.creditcode) where d.credit_code is not null");
            qcc_branch_update_code.createOrReplaceTempView("qcc_branch_update_code");

            spark.sql("select c.* from zs_branch c left join (select distinct a.record_id from qcc_branch_update_code a) d on c.record_id=d.record_id where d.record_id is null")
                    .createOrReplaceTempView("zs_branch_start_date");

            spark.sql("select " +
                            "a.record_id," +
                            "a.entid as id," +
                            "if(b.company_id is not null,b.company_id,md5(trim(a.name))) as key_no," +
                            "if(b.company_id is not null,b.company_id,md5(trim(a.name))) as company_id," +
                            "if(b.company_id is not null,b.company_name,a.name) as company_name," +
                            "if(length(a.BRN_CREDIT_CODE)==18,md5(trim(a.BRN_CREDIT_CODE)),concat(md5(trim(a.BRNAME)),'&')) as sub_key_no," +
                            "a.BRNAME AS name," +
                            "a.BRREGNO as reg_no," +
                            "a.BRN_REG_ORG as belong_org," +
                            "a.BRN_CREDIT_CODE," +
                            "a.jobid " +
                            "from zs_branch_start_date a " +
                            "left join " +
                            "a_dc_ep_dwi.t_eci_company b " +
                            "on regexp_replace(trim(a.name),'([(（）)])','') =regexp_replace(trim(b.company_name),'([(（）)])','') " +
                            "and regexp_replace(substr(cast(a.ESDATE as string),1,10),'([^0-9]+)','') = regexp_replace(substr(cast(b.start_date as string),1,10),'([^0-9]+)','') " +
                            "and trim(a.oper_name) = trim(b.oper_name)")
                    .createOrReplaceTempView("qcc_t_branch_tmp");

            spark.catalog().dropTempView("zs_branch_start_date");
            spark.sql("select * from (select a.* from qcc_branch_update_code a union all select b.* from qcc_t_branch_tmp b)")
                    .createOrReplaceTempView("qcc_t_br_first");
            //开始处理分支机构的相关数据
            spark.catalog().dropTempView("qcc_t_branch_tmp");
            spark.sql("select " +
                    "a.record_id," +
                    "a.id," +
                    "a.key_no," +
                    "a.company_id," +
                    "a.company_name," +
                    "a.sub_key_no," +
                    "if(b.company_id is not null,b.company_id,if(length(trim(a.BRN_CREDIT_CODE)) = 18,md5(trim(a.BRN_CREDIT_CODE)),md5(trim(a.name))))," +
                    "a.name," +
                    "a.reg_no," +
                    "a.belong_org," +
                    "a.jobid," +
                    "'0' as isadd " +
                    "from " +
                    "(select * from qcc_t_br_first where length(trim(BRN_CREDIT_CODE)) =18 ) a " +
                    "left join " +
                    "(select * from a_dc_ep_dwi.t_eci_company " +
                    "where length(trim(credit_code)) =18 ) b on trim(a.BRN_CREDIT_CODE)=trim(b.credit_code) where b.credit_code is not null")
                    .createOrReplaceTempView("sub_qcc_code");
            spark.sql("select c.* from qcc_t_br_first c left join (select distinct a.record_id from sub_qcc_code a) d on c.record_id=d.record_id where d.record_id is null")
                    .createOrReplaceTempView("zs_branch_sub_start_date");
            spark.sql("select " +
                    "a.record_id," +
                    "a.id," +
                    "a.key_no," +
                    "a.company_id," +
                    "a.company_name," +
                    "if(b.company_id is not null,b.company_id,if(length(trim(a.BRN_CREDIT_CODE)) = 18,concat(md5(trim(a.BRN_CREDIT_CODE)),'#'),concat(md5(trim(a.name)),'@'))) as sub_key_no," +
                    "if(b.company_id is not null,b.company_id,if(length(trim(a.BRN_CREDIT_CODE)) = 18,md5(trim(a.BRN_CREDIT_CODE)),md5(trim(a.name)))) as sub_company_id," +
                    "a.name," +
                    "a.reg_no," +
                    "a.belong_org," +
                    "a.jobid," +
                    "'0' as isadd " +
                    "from " +
                    "zs_branch_sub_start_date a " +
                    "left join " +
                    "a_dc_ep_dwi.t_eci_company b " +
                    "on regexp_replace(trim(a.name),'([(（）)])','')=regexp_replace(trim(b.company_name),'([(（）)])','')")
                    .createOrReplaceTempView("sub_qcc_code1");

            spark.catalog().dropTempView("qcc_t_br_first");
            spark.sql("select distinct * from (select * from sub_qcc_code1 union all select * from sub_qcc_code union all select * from qcc_branch_del)")
                    .drop("record_id").createOrReplaceTempView("last_tmp1");
            Dataset<Row> empty_id_rdd = spark.sql("select * from last_tmp1 where company_id is null");
            if(empty_id_rdd.count()>0){
                empty_id_rdd.createOrReplaceTempView("empty_id_branch");
                spark.sql("select a.id,if(a.company_name1!=a.name,b.key_no,null),if(a.company_name1!=a.name,b.company_id,null),if(a.company_name1!=a.name,b.company_name,null),a.sub_key_no,a.sub_company_id,a.name,a.reg_no,a.belong_org,a.jobid,a.isadd from \n" +
                        "(select *,if(size(split(name,\"公司\"))>=2,concat(split(name,\"公司\")[0],\"公司\"),if(size(split(name,\"中心\"))>=2,concat(split(name,\"中心\")[0],\"中心\"),name)) as company_name1\n" +
                        "from empty_id_branch) as a left join a_dc_ep_dwi.t_eci_company b \n" +
                        "on regexp_replace(trim(a.company_name1),'([(（）)])','')=regexp_replace(trim(b.company_name),'([(（）)])','')")
                        .createOrReplaceTempView("last_tmp2");
                spark.sql("insert overwrite table a_dc_ep_incr.t_eci_branch (select * from last_tmp1 where company_id is not null union all select * from last_tmp2)");
            }else {
                spark.sql("insert overwrite table a_dc_ep_incr.t_eci_branch select * from last_tmp1");
            }
            spark.catalog().dropTempView("sub_qcc_code");
            spark.catalog().dropTempView("sub_qcc_code1");
            spark.catalog().dropTempView("qcc_branch_del");
            //----------------------------------------------------------------------------------------------------
//            //com_company表是用来积累所有中数的对应的企业全量信息（不可以删，只可以加入）。
//            spark.sql("select b.* ,d.company_name as name from (select a.* from a_dc_ep_ods.zs_t_branch_del a where a.creditcode !='' and a.creditcode is not null) b left join (select distinct c.company_name ,c.credit_code from a_dc_ep_ods.t_eci_company c where " +
//                    "c.credit_code is not null and c.credit_code !='') d on b.creditcode = d.credit_code").union(spark.sql("select b.* ,d.name from (select a.* from a_dc_ep_ods.zs_t_branch_del a where a.creditcode ='' or a.creditcode is  null) b left join (select m.* from a_dc_ep_ods.com_company m where m.name !='' and m.name is not null) " +
//                    "d on b.entid = d.entid")).createOrReplaceTempView("zs_branch_del11");
//            spark.sql("select * from zs_branch_del11 where name is not null and name !=''").createOrReplaceTempView("zs_branch_del");
//            //中数分支机构表关联去统一社会信用代码 通过all_credit_code确定唯一
//            spark.sql("SELECT distinct a.RECORD_ID,a.entid,a.brname,a.brn_credit_code,a.brregno," +
//                    "       a.brn_reg_org,\n" +
//                    "       a.jobid,\n" +
//                    "       b.creditcode,\n" +
//                    "       md5(concat(regexp_replace(trim(b.name),'([(（）)])',''),regexp_replace(trim(a.brname),'([(（）)])',''))) AS all_credit_code," +
//                    "       b.name as entname \n" +
//                    "FROM a_dc_ep_ods.zs_t_branch a\n" +
//                    "LEFT JOIN zs_branch_del b ON a.entid = b.entid ").createOrReplaceTempView("zs_branch_tmp");
//        //企查查分支机构数据取母公司和分公司的统一社会信用代码 通过all_credit_code确定唯一
//        spark.sql("SELECT kk.*,\n" +
//                "          aa.credit_code AS sub_credit_code,\n" +
//                "          md5(concat(regexp_replace(trim(kk.company_name),'([(（）)])',''),regexp_replace(trim(aa.company_name),'([(（）)])',''))) AS all_credit_code\n" +
//                "   FROM\n" +
//                "     (SELECT c.*,\n" +
//                "             e.credit_code\n" +
//                "      FROM a_dc_ep_dwi.t_eci_branch c\n" +
//                "      LEFT JOIN\n" +
//                "        (SELECT company_id,\n" +
//                "                credit_code\n" +
//                "         FROM a_dc_ep_ods.t_eci_company\n" +
//                "         WHERE credit_code IS NOT NULL\n" +
//                "           AND credit_code !='') e ON c.company_id=e.company_id\n" +
//                "      WHERE e.company_id IS NOT NULL) kk\n" +
//                "   LEFT JOIN\n" +
//                "     a_dc_ep_ods.t_eci_company aa ON kk.sub_company_id=aa.company_id\n" +
//                "   WHERE aa.company_id IS NOT NULL ").createOrReplaceTempView("qcc_branch_tmp");
////            //企查查分支机构数据取母公司和分公司的统一社会信用代码 通过all_credit_code确定唯一
////            spark.sql("SELECT c.*, \n" +
////                    "          md5(concat(c.company_name,c.name)) AS all_credit_code\n" +
////                    "   FROM a_dc_ep_dwi.t_eci_branch c").createOrReplaceTempView("qcc_branch_tmp");
//            //更新数据
//            Dataset<Row> update_sql = spark.sql("select b.id,b.key_no,b.company_id,b.company_name,b.sub_key_no,b.sub_company_id,a.BRNAME as name,b.reg_no,b.belong_org,a.jobid,'0' as isadd from zs_branch_tmp a inner join qcc_branch_tmp b on a.all_credit_code = b.all_credit_code");
//            update_sql.createOrReplaceTempView("tmp1");
//
//            //新增数据
//            Dataset<Row> insert_sql = spark.sql("select a.RECORD_ID,a.entname,a.creditcode,a.brn_credit_code,a.brname ,a.BRREGNO,a.BRN_REG_ORG,a.jobid,'1' as isadd  from zs_branch_tmp a left join qcc_branch_tmp b on a.all_credit_code = b.all_credit_code where b.all_credit_code is null");
//            insert_sql.createOrReplaceTempView("tmp2");
//            spark.sql("select a.*,b.company_id from tmp2 a left join (select distinct d.company_id,d.company_name from a_dc_ep_ods.t_eci_company d) b on a.entname = b.company_name").createOrReplaceTempView("tmp4");
//            spark.sql("select a.*,b.company_id as sub_company_id from tmp4 a left join (select distinct d.company_id,d.company_name from a_dc_ep_ods.t_eci_company d) b on a.brname = b.company_name").createOrReplaceTempView("tmp5");
//            spark.sql("SELECT a.RECORD_ID,\n" +
//                    "       if(a.company_id IS NULL,md5(if(length(a.creditcode)=18,a.creditcode,a.entname)),a.company_id) AS key_no,\n" +
//                    "       if(a.company_id IS NULL,md5(if(length(a.creditcode)=18,a.creditcode,a.entname)),a.company_id) AS company_id,\n" +
//                    "       a.entname AS company_name,\n" +
//                    "       md5(if(length(a.brn_credit_code)=18,a.brn_credit_code,a.brname)) AS sub_key_no,\n" +
//                    "       md5(if(length(a.brn_credit_code)=18,a.brn_credit_code,a.brname)) AS sub_company_id,\n" +
//                    "       a.brname,\n" +
//                    "       a.BRREGNO,\n" +
//                    "       a.BRN_REG_ORG,\n" +
//                    "       a.jobid,\n" +
//                    "       a.isadd\n" +
//                    "FROM tmp5 a").createOrReplaceTempView("tmp6");
//
//            //删除数据
//            Dataset<Row> delete_sql = spark.sql("select a.id,a.key_no,a.company_id,a.company_name,a.sub_key_no,a.sub_company_id,a.name,a.reg_no,a.belong_org,b.jobid,'-1' as isadd from qcc_branch_tmp a inner join (select distinct del.* from zs_branch_del del left join a_dc_ep_ods.zs_t_branch k on del.entid = k.entid where k.entid is null and del.creditcode is not null and del.creditcode !='') b on a.credit_code = b.creditcode");
//            delete_sql.createOrReplaceTempView("tmp3");
//            spark.sql("select b.* from (select a.*,row_number() over(partition by a.company_name,a.name order by a.jobid) num from (select distinct c.* from tmp1 c union all select distinct d.* from tmp6 d union all select distinct e.* from tmp3 e) a ) b where b.num = 1").drop("num").createOrReplaceTempView("aaa");
//
//            spark.sql("insert overwrite table a_dc_ep_incr.t_eci_branch select * from aaa");
////            spark.sql("insert overwrite table a_dc_ep_incr.t_eci_branch select a.id,a.key_no,if(b.end_company_id is not null,b.end_company_id, a.company_id),a.company_name,a.sub_key_no,if(c.end_company_id is not null,c.end_company_id,a.sub_company_id ),a.name,a.reg_no,a.belong_org,a.jobid,a.isadd from aaa as a left join (select * from a_dc_ep_ods.clean_t_eci_company where length(company_id)>5 and length(end_company_id)>5) b on a.company_id=b.company_id left join a_dc_ep_ods.clean_t_eci_company c on a.sub_company_id=c.company_id");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_eci_branch");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
